package com.wjk.kylin.rpc.core.netty.client;

import com.wjk.kylin.rpc.core.invoker.reference.RpcReferenceBean;
import com.wjk.kylin.rpc.core.netty.AbstractClient;
import com.wjk.kylin.rpc.core.netty.params.RpcRequest;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 客户端
 * @author wjk
 */
public class NettyClient extends AbstractClient {
    private static volatile ConcurrentMap<String, NettyConnectClient> connectClientMap;
    private static final ConcurrentMap<String, Object> CONNECT_CLIENT_LOCK_MAP = new ConcurrentHashMap<>();

    /**
     * 执行请求
     *
     * @param address
     * @param rpcRequest
     * @throws Exception
     */
    @Override
    public void asyncSend(String address, RpcRequest rpcRequest) throws Exception {
        asyncSend(rpcRequest, address, rpcReferenceBean);
    }

    /**
     * async send
     */
    public static void asyncSend(RpcRequest rpcRequest, String address, RpcReferenceBean rpcReferenceBean) throws Exception {

        // client pool	[tips03 : may save 35ms/100invoke if move it to constructor, but it is necessary. cause by ConcurrentHashMap.get]
        NettyConnectClient connectClient = getNettyConnectClient(address, rpcReferenceBean);

        try {
            // do invoke
            connectClient.send(rpcRequest);
        } catch (Exception e) {
            LOGGER.error(">>>>>>>>>>> kylin-rpc clientPool send error :", e);
            throw e;
        }
    }

    /**
     * 获取客户端连接池
     *
     * @param address
     * @param rpcReferenceBean
     * @return
     * @throws Exception
     */
    private static NettyConnectClient getNettyConnectClient(String address, RpcReferenceBean rpcReferenceBean) throws Exception {

        // init base compont, avoid repeat init
        if (connectClientMap == null) {
            synchronized (NettyConnectClient.class) {
                if (connectClientMap == null) {
                    // init
                    connectClientMap = new ConcurrentHashMap<>(16);

                    // stop callback
                    rpcReferenceBean.getConsumerFactory().addStopCallBack(() -> {
                        if (connectClientMap.size() > 0) {
                            for (String key : connectClientMap.keySet()) {
                                NettyConnectClient clientPool = connectClientMap.get(key);
                                clientPool.close();
                            }
                            connectClientMap.clear();
                        }
                    });
                }
            }
        }

        // get-valid client
        NettyConnectClient connectClient = connectClientMap.get(address);
        if (connectClient != null && connectClient.isValidate()) {
            return connectClient;
        }

        // lock 防止多个请求 创建多个相同地址的连接对象
        Object clientLock = CONNECT_CLIENT_LOCK_MAP.get(address);
        if (clientLock == null) {
            CONNECT_CLIENT_LOCK_MAP.putIfAbsent(address, new Object());
            clientLock = CONNECT_CLIENT_LOCK_MAP.get(address);
        }

        // remove-create new client
        synchronized (clientLock) {

            // get-valid client, avlid repeat
            connectClient = connectClientMap.get(address);
            if (connectClient != null && connectClient.isValidate()) {
                return connectClient;
            }

            // remove old
            if (connectClient != null) {
                connectClient.close();
                connectClientMap.remove(address);
            }

            // set pool
            NettyConnectClient connectClientNew = new NettyConnectClient();
            try {
                connectClientNew.init(address, rpcReferenceBean.getSerializer(), rpcReferenceBean.getConsumerFactory());
                connectClientMap.put(address, connectClientNew);
            } catch (Exception e) {
                connectClientNew.close();
                throw e;
            }

            return connectClientNew;
        }

    }
}
